package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Code01SparkWordCount extends App {

  /**
   * 回顾MR:
   * 1.创建Configuration配置类对象
   * 2.创建Job操作类对象
   * 3.设置Job的名称
   * 4.设置计算逻辑(Map/Reduce)配置
   */

  // 1.如果要使用Spark进行数据计算，那么需要先设置Spark的执行环境
  private val conf: SparkConf = new SparkConf()
  conf.setAppName("WordCount")
  conf.setMaster("local") // 设置运行模式为本地模式

  private val sc = new SparkContext(conf) // 获取操作类的对象 -> 通常称为 sc

  // 2.获取操作对象后，再读取数据
  //   读取数据后返回的是一个RDD其中 数据格式泛型为String 类比 Scala中的List[String] 注意：RDD不是一个容器，并不会保存数据
  //   minPartitions 最小分区数据表示限定下限 那么对于切片数如果和最小分区数不匹配，那么会以倍数形式增加
  private val wordFileData: RDD[String] = sc.textFile("spark_code/data/word",3)
  println("wordFileDataRDDPartition:" + wordFileData.getNumPartitions)

  // 3.数据计算
  private val flatMapWord: RDD[String] = wordFileData
    //    .foreach(x => println("line:" + x))
    .flatMap(
      (line: String) => {
        // java,spark,java,hadoop
        println("flatmap执行了...")
        line.split(",")
      }

    )
  println("flatMapWordRDDPartition:" + flatMapWord.getNumPartitions)


  private val firstMap: RDD[String] = flatMapWord.map {
        //  [java,spark,java,hadoop] => 每个单词都会触发一次map计算逻辑
    case word => {
      println("first map running...")
      word
    }
  }




  private val groupByWord: RDD[(String, Iterable[String])] = firstMap.groupBy(x => x, 2)

  println("groupByWordRDDPartition:" + groupByWord.getNumPartitions)


  private val mapRes: RDD[(String, Int)] = groupByWord.map {
    case (word, wordLine) => {
      println("map执行了...")
      (word, wordLine.size)
    }
  }

  println("mapResRDDPartition:" + mapRes.getNumPartitions)

  // Job是由特定函数去触发的
  mapRes.foreach(println) // 通过两个foreach函数分别触发了两个job任务
  mapRes.foreach(println)


  // http://DESKTOP-9FB2DLF:4040
  // 使用while循环让当前代码处于运行状态，可以通过页面查看4040端口访问执行过程
  //  在执行过程中可以看到有两个job任务，在单个的job任务中存在有 DAG有向无环图
  //   可以在有向无环图中可以查看其血缘关系并且可以在图中查看 Stage,Stage表示对处理过程进行划分阶段
  //   在 Cluster Manager 的辅助下，把 task 任务分发调度出去
  //         那么Task是如何生成的？
  //             通过FileInputFormat类中的getSplit函数可以获取数据的切片，根据切片可以创建对应的Task(Partition)
  //             Task是在Executor中生成的，Executor是一个进程，Task是Executor的一个线程
  //

  /**
   * RDD：Resilient Distributed Dataset
   * RDD： 是一个弹性分布式数据集
   *
   *
   */

  while (true) {

  }


}
